# coding: utf-8
# Author: Legend奇
# @Time： 2020/8/13 上午11:29
import logging

from ngs import ssl as ngs_ssl
from ngs.tools.influxdb_client import InfluxDBClient
from ngs.tools.logs import DebugLog
from pyflux_influxdb.flux import FluxQuery, FluxQ, FluxR
from pyflux_influxdb.type_convert import flux_int

from settings import INFLUXDB_HOST, INFLUXDB_PORT, INFLUXDB_USER, INFLUXDB_PASSWORD, RUNNING_LOG_NAME, \
    INFLUXDB_DATABASE, INFLUXDB_RETENTION_POLICY

logger = logging.getLogger(RUNNING_LOG_NAME)


async def get_influx_conn():
    influx_client = InfluxDBClient(
        url=f"https://{INFLUXDB_HOST}:{INFLUXDB_PORT}",
        token=f'{INFLUXDB_USER}:{INFLUXDB_PASSWORD}',
        org='-',
        ssl_ca_cert=ngs_ssl.CA_FILE
    )
    return influx_client


async def influx_query_count(measurement, filters: list):
    flux_query = FluxQuery(f'{INFLUXDB_DATABASE}/{INFLUXDB_RETENTION_POLICY}').range('-1y')
    flux_query.filter(FluxQ(FluxR().__measurement__, measurement))
    tag_condition = None
    for field_item in filters:
        if field_item[1] and field_item[1] != 'None':
            if not tag_condition:
                tag_condition = FluxQ(getattr(FluxR(), field_item[0]), field_item[1])
            else:
                tag_condition = tag_condition.__and__(FluxQ(getattr(FluxR(), field_item[0]), field_item[1]))
    if tag_condition:
        flux_query.filter(tag_condition)
    script = flux_query.count('_value').get_script()
    influx_client = await get_influx_conn()
    tables = influx_client.query_api().query_stream(script)
    count = 0
    field = ''
    for record in tables:
        temp_field = record.values.get('_field')
        temp_value = record.values.get('_value')
        if not field:
            field = temp_field
            count += temp_value
        else:
            if temp_field == field:
                count += temp_value
    return count


async def get_number_flux(measurement, field, value):
    """
    from(bucket: "oplatform/autogen")
        |> range(start: -2m, stop: now())
        |> filter(fn: (r) => r._measurement == "partition.status" and r.device_id == "1")
        |> first()
        |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
        |> group(columns: ["_measurement"])
        |> count(column: "device_id")
    :param measurement: 表名
    :param field: 字段
    :param value: 值
    :return:
    """
    flux_query = FluxQuery(f'{INFLUXDB_DATABASE}/{INFLUXDB_RETENTION_POLICY}').range('-2y')
    if field and value:
        flux_query.filter(FluxQ(FluxR().__measurement__, measurement) & FluxQ(getattr(FluxR(), field), value))
    else:
        flux_query.filter(FluxQ(FluxR().__measurement__, measurement))
    flux_script = flux_query.first().pivot(['_time'], ['_field'], '_value').group(['_measurement']).count(
        'device_id').drop(['_measurement']).get_script()
    return await exec_query_by_flux(flux_script)[0].get('device_id')


async def get_flux_script(measurement, tag_filters, field_filters: list, limit, offset, order_field='_time', desc=False,
                          start='-2y', stop=None):
    """
    :param measurement: 表名
    :param tag_filters: tag字段过滤 [('device_id', '1')]
    :param field_filters: 其他字段过滤
    :param limit: 限制条数
    :param offset: 偏移量
    :param order_field: 排序字段
    :param desc: 是否倒序
    :param start: 开始时间
    :param stop: 结束时间
    :return:
    """
    flux_init = FluxQuery(f'{INFLUXDB_DATABASE}/{INFLUXDB_RETENTION_POLICY}')
    flux_init.range(start, stop).filter(FluxQ(FluxR().__measurement__, measurement))
    tag_condition = None
    if tag_filters:
        for tag_item in tag_filters:
            if not tag_condition:
                tag_condition = FluxQ(getattr(FluxR(), tag_item[0]), str(tag_item[1]))
            else:
                tag_condition = tag_condition.__or__(FluxQ(getattr(FluxR(), tag_item[0]), str(tag_item[1])))
    if tag_condition:
        flux_init.filter(tag_condition)
    flux_init \
        .pivot(['_time'], ['_field'], '_value') \
        .group(['_measurement']) \
        .sort([order_field], desc)
    flux_init \
        .map({'time': flux_int(flux_int(FluxR().__time__) + ' / 1000000')}) \
        .drop(['_measurement', '_time'])
    field_condition = None
    if field_filters:
        for field_item in field_filters:
            if not field_condition:
                field_condition = field_item
            else:
                field_condition = field_condition.__and__(field_item)
    if field_condition:
        flux_init.filter(field_condition)
    if limit > 0:
        flux_init.limit(limit, offset)
    return flux_init.get_script()


async def get_query_flux(measurement, tag_filters, conditions: list, limit, offset, order_field, desc, start='-2y',
                         stop=None):
    flux_script = get_flux_script(measurement, tag_filters, conditions, limit, offset, order_field, desc, start, stop)
    logger.debug(DebugLog(desc=f'操作命令为{flux_script}'))
    return await exec_query_by_flux(flux_script)


async def exec_query_by_flux(flux_script):
    try:
        influx_client = await get_influx_conn()
        tables = influx_client.query_api().query(flux_script)
        result = []
        for table in tables:
            for record in table.records:
                temp = record.values
                keys = temp.keys()
                if 'result' in keys:
                    del temp['result']
                if 'table' in keys:
                    del temp['table']
                if '_start' in keys:
                    del temp['_start']
                if '_stop' in keys:
                    del temp['_stop']
                result.append(temp)
        return result
    except:
        return []
